package repeater

import (
    "fmt"
    "github.com/influxdata/influxdb/client/v2"
    "time"
    "bytes"
    "io/ioutil"
)

func NewReportWriter(d map[string]string) (*NGReportWriter, error) {
    dbname := d["dbname"]
    user := d["user"]
    passwd := d["passwd"]
    addr := d["addr"]
    cacheFileName := d["cache_filename"]
    reportFileName := d["report_filename"]
    if cacheFileName == "" {
        return nil, fmt.Errorf("missing cache filename")
    }
    var influx client.Client
    var err error
    if dbname != "" && user != "" && passwd != "" && addr != "" {
        influx, err = client.NewHTTPClient(client.HTTPConfig{
            Addr:     addr,
            Username: user,
            Password: passwd,
            Timeout:  time.Second * 80,
        })
        if err != nil {
            return nil, err
        }
    }
    var filter *Filter
    filterPath := d["filter"]
    if filterPath != "" {
        f, err := LoadFilter(filterPath)
        if err != nil {
            return nil, err
        }
        filter = f
    }
    return &NGReportWriter{
        cacheFileName:  cacheFileName,
        dbName:         dbname,
        reportFileName: reportFileName,
        influx:         influx,
        StaticFunc:     ngStatic,
        localData:      make(map[int]DB),
        filter:         filter,
    }, nil
}

type NGReportWriter struct {
    localData      map[int]DB
    cacheFileName  string
    reportFileName string
    dbName         string
    StaticFunc     func(Value) map[string]float64
    workingOffset  int64
    commitedOffset int64
    filter         *Filter
    rc             *RepeaterController
    influx         client.Client
}

func (ww *NGReportWriter) newdb(day int) (db DB, err error) {
    fn := fmt.Sprintf("%s.%d", ww.cacheFileName, day)
    db = &LevelDB{desc: NewDBMeta(nil, day)}
    return db, db.Init(fn)
}

func (ww *NGReportWriter) Init() (err error) {
    ww.cacheFileName = fmt.Sprintf("%s.%d", ww.cacheFileName, ww.rc.Partition)
    if ww.reportFileName != "" {
        ww.reportFileName = fmt.Sprintf("%s.%d", ww.reportFileName, ww.rc.Partition)
    }
    return nil
}

func (ww *NGReportWriter) Start() {

}
func (ww *NGReportWriter) Stop() error {
    for _, v := range ww.localData {
        if err := v.Close(); err != nil {
            return err
        }
    }
    return ww.influx.Close()
}

func (ww *NGReportWriter) Name() string {
    return "report_ng"
}

func (ww *NGReportWriter) SetController(rc *RepeaterController) error {
    ww.rc = rc
    return nil
}

const (
    CloseTimeLimit  = 3600 * 4
    ReportSizeLimit = 300
    RemoveTimeLimit = 3600 * 24 * 3
)

func (ww *NGReportWriter) Commit() (int64, error) {
    toReport, err := ww.checkToReport()
    if err != nil {
        return ww.commitedOffset, err
    }
    rpts, err := ww.Report(toReport)
    if err != nil {
        return ww.commitedOffset, err
    }
    if len(rpts) > 0 {
        if err := ww.LocalSend(rpts); err != nil {
            return ww.commitedOffset, err
        }
        if err := ww.InfluxSend(rpts); err != nil {
            return ww.commitedOffset, err
        }
    }
    if err := ww.clean(toReport); err != nil {
        return ww.commitedOffset, err
    }
    for k, v := range ww.localData {
        if v.ShouldClose(CloseTimeLimit) {
            if err := v.Close(); err != nil {
                return ww.commitedOffset, err
            }
            log.Infof("close %d", k)
        }
        if v.ShouldDrop(RemoveTimeLimit) {
            if err := v.Drop(); err != nil {
                return ww.commitedOffset, err
            }
            delete(ww.localData, k)
            log.Infof("drop %d", k)
        }
    }
    var days []int
    for k := range ww.localData {
        days = append(days, k)
    }
    for _, day := range days {
        if err := ww.localData[day].Commit(); err != nil {
            return ww.commitedOffset, err
        }
    }
    log.Infof("working on dates: %v", days)
    ww.commitedOffset = ww.workingOffset
    return ww.commitedOffset, nil
}

func (ww *NGReportWriter) clean(reported map[int][]string) error {
    for k, v := range reported {
        for _, ip := range v {
            if err := ww.localData[k].Clean(ip); err != nil {
                return err
            }
        }
        if err := ww.localData[k].Commit(); err != nil {
            return err
        }
    }
    return nil
}

func (ww *NGReportWriter) Report(reportCandidates map[int][]string) ([]Report, error) {
    var ret []Report
    for day, ips := range reportCandidates {
        for _, ip := range ips {
            ds, err := ww.localData[day].ValuesFor(ip)
            if err != nil {
                return nil, err
            }
            ret = append(ret, ww.report(day, ip, ds)...)
        }
    }
    return ret, nil
}

func (ww *NGReportWriter) report(day int, ip string, set DataSet) (rpts []Report) {
    for k, v := range set {
        rpt := Report{ip, day, k, ww.StaticFunc(v)}
        rpts = append(rpts, rpt)
    }
    return
}

func (ww *NGReportWriter) checkToReport() (map[int][]string, error) {
    ipForDay := make(map[string][]int)
    for day, m := range ww.localData {
        ips, err := m.Range()
        if err != nil {
            return nil, err
        }
        for _, ip := range ips {
            switch {
            case len(ipForDay[ip]) == 0:
                ipForDay[ip] = []int{day}
            case ipForDay[ip][0] < day:
                ipForDay[ip] = append([]int{day}, ipForDay[ip]...)
            default:
                ipForDay[ip] = append(ipForDay[ip], day)
            }
        }
    }
    dayForIP := make(map[int][]string)
    for ip, days := range ipForDay {
        if len(days) == 1 {
            continue
        }
        for i, day := range days {
            if i == 0 {
                continue
            }
            dayForIP[day] = append(dayForIP[day], ip)
        }
    }
    return dayForIP, nil
}

func (ww *NGReportWriter) LocalSend(reports []Report) error {
    if ww.reportFileName == "" || len(reports) == 0 {
        return nil
    }
    fn := fmt.Sprintf("%s.%d", ww.reportFileName, time.Now().Unix())
    buf := new(bytes.Buffer)
    for _, r := range reports {
        buf.Write(r.toByte())
        buf.WriteString("\n")
    }
    return ioutil.WriteFile(fn, buf.Bytes(), 0644)
}

func (ww *NGReportWriter) InfluxSend(reports []Report) error {
    if ww.influx == nil {
        return nil
    }
    l := len(reports)
    if l < ReportSizeLimit {
        return ww.influxSend(reports)
    }
    for i := 0; i < l; i += ReportSizeLimit {
        next := i + ReportSizeLimit
        if i+ReportSizeLimit > l {
            next = l - 1
        }
        if err := ww.influxSend(reports[i:next]); err != nil {
            return err
        }
    }
    return nil
}

func (ww *NGReportWriter) influxSend(reports []Report) error {
    log.Infof("%d Report to send", len(reports))
    bp, err := client.NewBatchPoints(client.BatchPointsConfig{
        Database:  ww.dbName,
        Precision: "h",
    })
    if err != nil {
        return err
    }
    for _, rpt := range reports {
        tags := make(map[string]string)
        tags["ip"] = rpt.Ip
        tags["par"] = fmt.Sprintf("%d", ww.rc.Partition)
        tags["report_ts"] = fmt.Sprintf("%d", time.Now().Unix())
        if p := rpt.Metic.Path; p != "" {
            tags["path"] = p
        }
        fields := make(map[string]interface{})
        for k, v := range rpt.Data {
            fields[rpt.Metic.Sub+"_"+k] = v
        }
        ts, err := time.Parse("20060102", fmt.Sprintf("%d", rpt.Day))
        if err != nil {
            log.Warningf("ilegal datapoint %v", rpt.Day)
            continue
        }
        pt, err := client.NewPoint(rpt.Metic.Head, tags, fields, ts)
        bp.AddPoint(pt)
    }
    if err := ww.influx.Write(bp); err != nil && !IsBeyondRetention(err) {
        return err
    }
    return nil
}

func (ww *NGReportWriter) Handle(dps *DataPointPack) error {
    ww.workingOffset = dps.Offset()
    var days []int
    abnormalIP := make(map[string]interface{})
    for i, dp := range dps.Datapoints {
        day, ip, rm, v := parse(dp)
        if ww.filter != nil && !ww.filter.pass(dp.Name) {
            continue
        }
        if ip == "" || !rm.valid() {
            abnormalIP[ip] = new(interface{})
            continue
        }
        days = append(days, day)
        offset := dps.Offsets[i]
        if _, ok := ww.localData[day]; !ok {
            db, err := ww.newdb(day)
            if err != nil {
                return err
            }
            ww.localData[day] = db
        }
        ww.localData[day].Prepare(ip, offset, rm, v)
    }
    log.Warningf("abnormal ip: %v", abnormalIP)
    return nil
}
